use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, Condvar};
use tokio::sync::{mpsc, Mutex as TokioMutex};
use tokio::time::{timeout, Duration};
use serde::Serialize;
use serde_json;
use anyhow::{Result, anyhow};
use crate::bridge::arrow::{ArrowSerializer, DataType, BridgeError};
use arrow::datatypes::{Schema, Field};
use std::time::SystemTime;
use std::pin::Pin;
use std::future::Future;

#[derive(Debug)]
pub struct MetricsCollector {
    start_time: SystemTime,
    call_count: Mutex<u64>,
    error_count: Mutex<u64>,
}

#[derive(Debug, Serialize)]
pub struct MetricsReport {
    pub uptime_seconds: u64,
    pub total_calls: u64,
    pub error_count: u64,
}

impl MetricsCollector {
    pub fn new() -> Self {
        Self {
            start_time: SystemTime::now(),
            call_count: Mutex::new(0),
            error_count: Mutex::new(0),
        }
    }

    pub fn start_call(&self, _function_name: &str) -> SystemTime {
        *self.call_count.lock().unwrap() += 1;
        SystemTime::now()
    }

    pub fn record_error(&self, _function_name: &str) {
        *self.error_count.lock().unwrap() += 1;
    }

    pub fn get_metrics_report(&self) -> MetricsReport {
        let uptime = SystemTime::now()
            .duration_since(self.start_time)
            .unwrap_or_default()
            .as_secs();
        
        MetricsReport {
            uptime_seconds: uptime,
            total_calls: *self.call_count.lock().unwrap(),
            error_count: *self.error_count.lock().unwrap(),
        }
    }
}

pub struct Task {
    id: u64,
    future: Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send>>> + Send>>,
}

impl Task {
    pub fn new(id: u64, future: Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send>>> + Send>>) -> Self {
        Self { id, future }
    }
}

pub struct TaskQueue {
    tasks: TokioMutex<VecDeque<Task>>,
}

impl TaskQueue {
    pub fn new() -> Self {
        Self {
            tasks: TokioMutex::new(VecDeque::new()),
        }
    }

    pub async fn push(&self, task: Task) -> Result<(), Box<dyn std::error::Error + Send>> {
        let mut tasks = self.tasks.lock().await;
        tasks.push_back(task);
        Ok(())
    }

    pub async fn pop(&self) -> Option<Task> {
        let mut tasks = self.tasks.lock().await;
        tasks.pop_front()
    }
}

type CallbackFn = Box<dyn Fn(Vec<u8>) -> Result<Vec<u8>> + Send + Sync>;

/// 异步桥接管理器
pub struct AsyncBridge {
    /// 注册的函数映射
    functions: Arc<Mutex<HashMap<String, CallbackFn>>>,
    /// 事件发送通道
    event_tx: mpsc::Sender<(String, Vec<u8>)>,
    /// 事件接收通道
    event_rx: Arc<Mutex<Option<mpsc::Receiver<(String, Vec<u8>)>>>>,
    /// 性能指标收集器
    metrics: MetricsCollector,
}

impl AsyncBridge {
    /// 创建新的异步桥接实例
    pub fn new() -> Self {
        let (tx, rx) = mpsc::channel(100);
        
        Self {
            functions: Arc::new(Mutex::new(HashMap::new())),
            event_tx: tx,
            event_rx: Arc::new(Mutex::new(Some(rx))),
            metrics: MetricsCollector::new(),
        }
    }
    
    /// 注册可以被Python调用的函数
    pub fn register_function<F>(&self, name: &str, callback: F) 
    where
        F: Fn(Vec<u8>) -> Result<Vec<u8>> + Send + Sync + 'static,
    {
        let mut functions = self.functions.lock().unwrap();
        functions.insert(name.to_string(), Box::new(callback));
    }
    
    /// 获取性能指标报告
    pub fn get_metrics_report(&self) -> serde_json::Value {
        serde_json::to_value(self.metrics.get_metrics_report()).unwrap_or_default()
    }
    
    /// 处理来自Python的调用
    pub async fn handle_call(&self, function_name: &str, data: Vec<u8>, timeout_ms: Option<u64>) -> Result<Vec<u8>> {
        let _tracker = self.metrics.start_call(function_name);
        let functions = self.functions.lock().unwrap();
        
        // 查找注册的函数
        if let Some(callback) = functions.get(function_name) {
            // 如果设置了超时
            if let Some(ms) = timeout_ms {
                match timeout(Duration::from_millis(ms), async {
                    // 在另一个线程中执行回调，避免阻塞异步运行时
                    let callback = callback.clone();
                    let data = data.clone();
                    
                    tokio::task::spawn_blocking(move || {
                        callback(data)
                    }).await.unwrap()
                }).await {
                    Ok(result) => result,
                    Err(_) => {
                        // 记录错误并返回超时错误
                        self.metrics.record_error(function_name);
                        let error = BridgeError {
                            error_code: 408,
                            error_type: "Timeout".to_string(),
                            message: format!("Function call '{}' timed out after {}ms", function_name, ms),
                            details: None,
                        };
                        
                        let schema = Schema::new(vec![
                            Field::new("error_code", arrow::datatypes::DataType::Int32, false),
                            Field::new("error_type", arrow::datatypes::DataType::Utf8, false),
                            Field::new("message", arrow::datatypes::DataType::Utf8, false),
                            Field::new("details", arrow::datatypes::DataType::Utf8, true),
                        ]);
                        let serializer = ArrowSerializer::new(schema);
                        Ok(serializer.to_json_bytes(&error, DataType::Error)?)
                    }
                }
            } else {
                // 无超时的调用
                let callback = callback.clone();
                let data = data.clone();
                
                tokio::task::spawn_blocking(move || {
                    callback(data)
                }).await?
            }
        } else {
            // 记录错误并返回函数未找到错误
            self.metrics.record_error(function_name);
            let error = BridgeError {
                error_code: 404,
                error_type: "NotFound".to_string(),
                message: format!("Function '{}' not found", function_name),
                details: None,
            };
            
            let schema = Schema::new(vec![
                Field::new("error_code", arrow::datatypes::DataType::Int32, false),
                Field::new("error_type", arrow::datatypes::DataType::Utf8, false),
                Field::new("message", arrow::datatypes::DataType::Utf8, false),
                Field::new("details", arrow::datatypes::DataType::Utf8, true),
            ]);
            let serializer = ArrowSerializer::new(schema);
            Ok(serializer.to_json_bytes(&error, DataType::Error)?)
        }
    }
    
    /// 发送事件到Python
    pub async fn send_event(&self, event_type: &str, data: Vec<u8>) -> Result<()> {
        self.event_tx.send((event_type.to_string(), data)).await
            .map_err(|e| anyhow!("Failed to send event: {}", e))
    }
    
    /// 获取事件接收器
    pub fn take_event_receiver(&self) -> Option<mpsc::Receiver<(String, Vec<u8>)>> {
        let mut rx_guard = self.event_rx.lock().unwrap();
        rx_guard.take()
    }
    
    /// 安全调用，捕获异常并转换为错误响应
    pub fn safe_call<F, T>(&self, func: F) -> Vec<u8>
    where
        F: FnOnce() -> Result<T>,
        T: Serialize,
    {
        let serializer = ArrowSerializer::new(Default::default());
        
        match func() {
            Ok(result) => {
                match serializer.to_json_bytes(&result, DataType::MarketData) {
                    Ok(bytes) => bytes,
                    Err(e) => {
                        let error = BridgeError {
                            error_code: 500,
                            error_type: "SerializationError".to_string(),
                            message: format!("Failed to serialize result: {}", e),
                            details: None,
                        };
                        
                        serializer.to_json_bytes(&error, DataType::Error)
                            .unwrap_or_else(|_| b"Internal error".to_vec())
                    }
                }
            },
            Err(e) => {
                let error = BridgeError {
                    error_code: 500,
                    error_type: "ExecutionError".to_string(),
                    message: format!("{}", e),
                    details: None,
                };
                
                serializer.to_json_bytes(&error, DataType::Error)
                    .unwrap_or_else(|_| b"Internal error".to_vec())
            }
        }
    }
}

impl Default for AsyncBridge {
    fn default() -> Self {
        Self::new()
    }
}